package com.example.kafka.task;

import com.example.kafka.constant.KafkaConstant;
import com.example.kafka.constant.StreamConstant;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.SendResult;
import org.springframework.scheduling.annotation.EnableScheduling;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;
import org.springframework.util.concurrent.ListenableFuture;
import org.springframework.util.concurrent.ListenableFutureCallback;

import java.util.stream.IntStream;

/**
 * @application:
 * @author: LiYuhang
 * @Date: 2021/6/26 20:31
 * @version: 0.1
 * @Description :
 * @Modified By:
 */
@Slf4j
@Component
@EnableScheduling
public class StreamProducer {

    private KafkaTemplate<String, Object> kafkaTemplate;

    public KafkaTemplate<String, Object> getKafkaTemplate() {
        return kafkaTemplate;
    }
    @Autowired
    public void setKafkaTemplate(KafkaTemplate<String, Object> kafkaTemplate) {
        this.kafkaTemplate = kafkaTemplate;
    }

    @Scheduled(cron = "0/10 * * * * ? ")
    public void messageProducer() {
        IntStream.range(0,KafkaConstant.ONCE_SEND_SIZE).forEach(v -> {
            String key;
            if (v % 2 == 0) {
                key = "00001" + v;
            } else {
                key = "00002" + v;
            }
            ListenableFuture<SendResult<String, Object>> future = kafkaTemplate
                    .send(StreamConstant.IN_PUT_TOPIC,key,KafkaConstant.generatorUuid());
            future.addCallback(new ListenableFutureCallback<SendResult<String, Object>>() {
                @Override
                public void onFailure(Throwable throwable) {
                    //发送失败的处理
                    log.error("发送失败，{}",throwable.getMessage(),throwable);
                }

                @Override
                public void onSuccess(SendResult<String, Object> stringObjectSendResult) {
                    //成功的处理
                    log.info(" - 生产者 发送消息成功：" + stringObjectSendResult.toString());
                }
            });
        });

    }
}
